package com.galen.subscriber.server;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.galen.subscriber.event.CanalEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author shuaiys
 * @version 1.0
 * @package com.galen.subscriber.server
 * @description TODO
 * @date 2020-03-08 15:53
 */
@Component
@Slf4j
public class SubscriberServer {

    @Resource
    private CanalConnector connector;

    @Resource
    private ApplicationEventPublisher publisher;

    // 单次从canal server获取的数量
    private final static Integer MSG_SIZE = 100;

    public void run() {
        while (true) {
            Message message = connector.getWithoutAck(MSG_SIZE);
            long batchId = message.getId();
            try {
                List<CanalEntry.Entry> entries = message.getEntries();
                // 空集合直接跳过
                if (batchId != -1L && entries.size() > 0) {
                    entries.forEach(entry -> {
                        if (entry.getEntryType().getNumber() == CanalEntry.EntryType.ROWDATA.getNumber()){
                            publisher.publishEvent(new CanalEvent(entry));
                        }
                    });
                }
                connector.ack(batchId);
            } catch (Exception e) {
                connector.rollback(batchId);
                log.info("发生异常，canal batchId 回滚， batchId={}", batchId);
            }

            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                connector.disconnect();
                log.info("发生异常,退出canal连接...");
            }
        }
    }

}